package ModelCreator;


import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.cassandra.service.ThriftKsDef;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.ColumnSlice;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.ddl.ColumnFamilyDefinition;
import me.prettyprint.hector.api.ddl.ComparatorType;
import me.prettyprint.hector.api.ddl.KeyspaceDefinition;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.mutation.Mutator;
import me.prettyprint.hector.api.query.QueryResult;
import me.prettyprint.hector.api.query.SliceQuery;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class WorkflowIterator {

    private Cluster cluster;
    private ColumnFamilyDefinition columnFamilyDefinition;
    private KeyspaceDefinition newKeyspace;
    private Keyspace keyspace;

    public WorkflowIterator() {
        cluster = HFactory.getOrCreateCluster(ModelConfig.CASSANDRA_CLUSTER, ModelConfig.CLUSTER_PORT);
        columnFamilyDefinition = HFactory.createColumnFamilyDefinition(ModelConfig.KEYSPACE, ModelConfig.WORKFLOW_ITERATOR_COLUMN_FAMILY, ComparatorType.BYTESTYPE);
        if (cluster.describeKeyspace(ModelConfig.KEYSPACE) == null) {
            newKeyspace = HFactory.createKeyspaceDefinition(ModelConfig.KEYSPACE, ThriftKsDef.DEF_STRATEGY_CLASS, 1, Arrays.asList(columnFamilyDefinition));
            cluster.addKeyspace(newKeyspace, true);
        } else
            cluster.addColumnFamily(columnFamilyDefinition);
        keyspace = HFactory.createKeyspace(ModelConfig.KEYSPACE, cluster);
        //cluster.addColumnFamily(columnFamilyDefinition);

    }

    /**
     * @param workflowID
     * @param timestamp
     * @param status
     */
    public void createWorkflowIterator(String workflowID, String timestamp, String status) {

        Mutator<String> mutator = HFactory.createMutator(keyspace, StringSerializer.get());

        mutator.addInsertion(status, ModelConfig.WORKFLOW_ITERATOR_COLUMN_FAMILY,
                HFactory.createStringColumn(timestamp + "_" + (int) (Math.random() * 9999), workflowID));
        mutator.execute();


    }

    public List<String> getWorkflowsForATimeDuration(String rowKey, String timestamp1, String timestamp2) {
        String newTimestamp1 = timestamp1.concat("_0000");
        String newTimestamp2 = timestamp2.concat("_9999");
        List<String> resultList = new ArrayList<String>();

        SliceQuery<String, String, String> query = HFactory.createSliceQuery(keyspace, StringSerializer.get(), StringSerializer.get(), StringSerializer.get());
        query.setColumnFamily(ModelConfig.WORKFLOW_ITERATOR_COLUMN_FAMILY).setKey(rowKey).setRange(newTimestamp1, newTimestamp2, false, 100);
        QueryResult<ColumnSlice<String, String>> result = query.execute();

        for (HColumn<String, String> column : result.get().getColumns()) {
            resultList.add(column.getValue());
            //System.out.println(column.getName() +"::"+ column.getValue());
        }
        return resultList;
    }


}